package com.book.configure;

import com.book.filter.CustomGateWayFilter;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import feign.codec.Decoder;
import org.apache.commons.compress.utils.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.ObjectFactory;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.http.HttpMessageConverters;
import org.springframework.cloud.openfeign.support.ResponseEntityDecoder;
import org.springframework.cloud.openfeign.support.SpringDecoder;
import org.springframework.context.annotation.Bean;
import org.springframework.http.MediaType;
import org.springframework.http.converter.HttpMessageConverter;
import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Description：
 *
 * @Author： leo.xiong
 * @CreateDate： 2020/12/03 15:17
 * @Email： leo.xiong@suyun360.com
 * @Since：
 */
@SpringBootConfiguration
public class GateWayConfig {

    @Value("${thread.max.pool.size:20}")
    private Integer threadMaxPoolSize;

    @Value("${thread.queue.capacity.size:20}")
    private Integer threadQueueCapacitySize;

    @Bean
    public Decoder feignDecoder() {
        return new ResponseEntityDecoder(new SpringDecoder(feignHttpMessageConverter()));
    }

    public ObjectFactory<HttpMessageConverters> feignHttpMessageConverter() {
        final HttpMessageConverters httpMessageConverters = new HttpMessageConverters(new GateWayMappingJackson2HttpMessageConverter());
        return new ObjectFactory<HttpMessageConverters>() {
            @Override
            public HttpMessageConverters getObject() throws BeansException {
                return httpMessageConverters;
            }
        };
    }

    public class GateWayMappingJackson2HttpMessageConverter extends MappingJackson2HttpMessageConverter {
        GateWayMappingJackson2HttpMessageConverter() {
            List<MediaType> mediaTypes = Lists.newArrayList();
            mediaTypes.add(MediaType.valueOf(MediaType.TEXT_HTML_VALUE + ";charset=UTF-8"));
            setSupportedMediaTypes(mediaTypes);
        }
    }

    @Bean
    public ExecutorService customGateWayFilterExecutorService() {
        Logger logger = LoggerFactory.getLogger(CustomGateWayFilter.class);
        ExecutorService executorService = buildExecutorService(logger, threadMaxPoolSize, threadQueueCapacitySize, "customGateWayFilter");
        return executorService;
    }

    @Bean
    @ConditionalOnMissingBean
    public HttpMessageConverters messageConverters(ObjectProvider<HttpMessageConverter<?>> converters) {
        return new HttpMessageConverters(converters.orderedStream().collect(Collectors.toList()));
    }

    /**
     * 新建线程池
     *
     * @param logger
     * @param threadMaxPoolSize
     * @param threadQueueCapacitySize
     * @param poolName
     * @return
     */
    private ExecutorService buildExecutorService(Logger logger, int threadMaxPoolSize, int threadQueueCapacitySize, String poolName) {
        return new ThreadPoolExecutor(threadMaxPoolSize, threadMaxPoolSize,
                0L, TimeUnit.MILLISECONDS,
                new PriorityBlockingQueue<>(threadQueueCapacitySize),
                new ThreadFactoryBuilder().setNameFormat("pool-" + poolName + "-consume-thread-%d").build(),
                new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        try {
                            long startTime = System.currentTimeMillis();
                            executor.getQueue().put(r);
                            logger.info("too many terminal data requests to handle,blocking the subscribe message {}ms", System.currentTimeMillis() - startTime);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                });
    }
}
